package com.jintian.smart.kernel.switching.dto;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufHolder;
import io.netty.buffer.CompositeByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.util.IllegalReferenceCountException;
import io.netty.util.ReferenceCountUtil;
import io.netty.util.ReferenceCounted;
import io.netty.util.internal.StringUtil;
import org.apache.commons.io.IOUtils;
import org.apache.dubbo.common.serialize.hessian2.Hessian2ObjectInput;
import org.apache.dubbo.common.serialize.hessian2.Hessian2ObjectOutput;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.*;
import java.util.HashMap;
import java.util.Map;

public class TransferMessage implements ITransferMessage, ByteBufHolder, Closeable {
    private final static Logger logger = LoggerFactory.getLogger(TransferMessage.class);
    private final static short[] HEADER_OFFSET = new short[]{0, 2, 4, 8};
    private final static short HEADER_VER_IDX = 0;
    private final static short HEADER_TYPE_IDX = 1;
    private final static short HEADER_TOTALLEN_IDX = 2;
    private final static short HEADER_ATTACHMENTLEN_IDX = 3;
    /**
     * 存放该帧数据的长度的数据的起始位置 0-1:msg_version,2-3:msg_type
     */
    public static final int LENGTH_FIELD_OFFSET = HEADER_OFFSET[HEADER_TOTALLEN_IDX];

    /**
     * 消息头部的总长度
     */
    public static final int HEADER_LENGTH = /* version */2 + /* msgTyppe */2 + /* msgLength */4
            + /* attachmentLength */ +4;
    private final HeaderWrapper header;

    private ByteBuf content;

    private TransferMessage(ByteBuf content) {
        this.header = new HeaderWrapper();
        this.content = content;
    }

    private Map<String, Object> attachment;

    @SuppressWarnings("unchecked")
    public Map<String, Object> attachment() {
        if (this.attachment == null) {
            if (this.header.getAttachmentSize() > 0) {
                byte[] dst = new byte[this.header.getAttachmentSize()];
                this.content.getBytes(this.header.getMsgSize() - this.header.getAttachmentSize(), dst);
                this.attachment = TransferMessage.deserialize(dst, Map.class);
            } else {
                this.attachment = new HashMap<>();
            }
        }
        return this.attachment;
    }

    public ByteBuf getContent() {
        return this.content;
    }

    public ByteBuf getMessageBodyBuf() {
        return this.content.slice(HEADER_LENGTH,
                this.getHeader().getMsgSize() - HEADER_LENGTH - this.getHeader().getAttachmentSize());
    }

    public void updateAttachment() {
        if (this.attachment == null) {
            // 没用进行附件修改
            return;
        }
        byte[] newAttachmentBuff = this.attachment.isEmpty() ? new byte[0] : TransferMessage.serialize(this.attachment);
        if (this.header.getAttachmentSize() > 0 && newAttachmentBuff.length == 0) {
            // 截断附件部分
            this.content = this.content.slice(0, this.header.getMsgSize() - this.header.getAttachmentSize());
        } else if (newAttachmentBuff.length <= this.header.getAttachmentSize()) {
            // 截断多余部分
            this.content = this.content.slice(0,
                    this.header.getMsgSize() - this.header.getAttachmentSize() + newAttachmentBuff.length);
            this.content.setBytes(this.header.getMsgSize() - this.header.getAttachmentSize(), newAttachmentBuff);
        } else if (newAttachmentBuff.length > this.header.getAttachmentSize()) {
            // 先截断附件，添加新的附件
            CompositeByteBuf newContent = this.content.alloc().compositeBuffer();
            newContent.addComponent(true,
                    this.content.slice(0, this.header.getMsgSize() - this.header.getAttachmentSize()));
            newContent.addComponent(true, Unpooled.wrappedBuffer(newAttachmentBuff));
            this.content = newContent;
        }
        this.header.updateAttachmentSize(newAttachmentBuff.length);
    }

    public IHeader getHeader() {
        return this.header;
    }

    class HeaderWrapper implements IHeader {
        @Override
        public MessageType getMessageType() {
            return MessageType.values()[content.getShort(HEADER_OFFSET[HEADER_TYPE_IDX])];
        }

        @Override
        public short getMsgVersion() {
            return content.getShort(HEADER_OFFSET[HEADER_VER_IDX]);
        }

        @Override
        public int getMsgSize() {
            return content.getInt(HEADER_OFFSET[HEADER_TOTALLEN_IDX]);
        }

        @Override
        public int getAttachmentSize() {
            return content.getInt(HEADER_OFFSET[HEADER_ATTACHMENTLEN_IDX]);
        }

        private void updateAttachmentSize(int newSize) {
            content.setInt(HEADER_OFFSET[HEADER_TOTALLEN_IDX], this.getMsgSize() - this.getAttachmentSize() + newSize);
            content.setInt(HEADER_OFFSET[HEADER_ATTACHMENTLEN_IDX], newSize);
        }

    }

    @Override
    public ByteBuf content() {
        return content;
    }

    @Override
    public int refCnt() {
        return content.refCnt();
    }

    @Override
    public TransferMessage retain() {
        content.retain();
        return this;
    }

    @Override
    public TransferMessage retain(int increment) {
        content.retain(increment);
        return this;
    }

    @Override
    public TransferMessage touch() {
        content.touch();
        return this;
    }

    @Override
    public TransferMessage touch(Object hint) {
        content.touch(hint);
        return this;
    }

    @Override
    public boolean release() {
        return content.release();
    }

    @Override
    public boolean release(int decrement) {
        return content.release(decrement);
    }

    @Override
    public String toString() {
        return StringUtil.simpleClassName(this) + "(data: " + content() + ')';
    }

    @Override
    public TransferMessage copy() {
        return replace(content.copy());
    }

    @Override
    public TransferMessage duplicate() {
        return replace(content.duplicate());
    }

    @Override
    public TransferMessage retainedDuplicate() {
        return replace(content.retainedDuplicate());
    }

    @Override
    public TransferMessage replace(ByteBuf content) {
        return new TransferMessage(content);
    }

    @Override
    public void close() throws IOException {
        this.release();
    }

    /**
     * Used to cache the value of the hash code and avoid
     * {@link IllegalReferenceCountException}.
     */
    private int hash;

    @Override
    public int hashCode() {
        int hash = this.hash;
        if (hash == 0) {
            if (content().refCnt() != 0) {
                try {
                    hash = 31 + content().hashCode();
                } catch (IllegalReferenceCountException ignored) {
                    // Handle race condition between checking refCnt() == 0 and
                    // using the object.
                    hash = 31;
                }
            } else {
                hash = 31;
            }
            hash = 31 * hash + attachment().hashCode();
            hash = 31 * hash + super.hashCode();
            this.hash = hash;
        }
        return hash;
    }

    private static <T> T deserialize(byte[] data, Class<T> clazz) {
        Hessian2ObjectInput hessian2Input = new Hessian2ObjectInput(new ByteArrayInputStream(data));
        try {
            return clazz.cast(hessian2Input.readObject(clazz));
        } catch (Exception e) {
            throw new IllegalArgumentException("deserialize " + clazz.getName() + " error", e);
        }
    }

    private static byte[] serialize(Object object) {
        if (object == null) {
            return new byte[0];
        } else {
            ByteArrayOutputStream out = new ByteArrayOutputStream();
            Hessian2ObjectOutput writer = new Hessian2ObjectOutput(out);
            try {
                writer.writeObject(object);
                writer.flushBuffer();
                return out.toByteArray();
            } catch (IOException e) {
                throw new IllegalArgumentException("serialize " + object.getClass().getName() + " error", e);
            }
        }
    }

    private static void writeHeader(ByteBuf headerBuff, short version, MessageType type) {
        // 版本 2 byte
        headerBuff.writeShort(version);
        // 数据类型 2 byte
        headerBuff.writeShort(type.ordinal());
        // 消息长度 4 byte
        headerBuff.writeInt(0);
        // 其中附件长度 4 byte
        headerBuff.writeInt(0);
    }

    private static void writeAttachment(CompositeByteBuf messageBuf, ByteBuf header, Map<String, Object> attachment) {
        // attachment
        if (attachment != null && !attachment.isEmpty()) {
            ByteBuf attachBuf = Unpooled.wrappedBuffer(TransferMessage.serialize(attachment));
            messageBuf.addComponent(true, attachBuf);
            // reset attachment size
            header.setInt(HEADER_OFFSET[HEADER_ATTACHMENTLEN_IDX], attachBuf.readableBytes());
        }
    }

    public static TransferMessage loadFromBuffer(ByteBuf messageBuff) {
        return new TransferMessage(messageBuff);
    }

    public static TransferMessage loadFromFile(File file) throws IOException {
        try (InputStream in = new FileInputStream(file)) {
            ByteBuf buf = Unpooled.copiedBuffer(IOUtils.toByteArray(in));
            return TransferMessage.loadFromBuffer(buf);
        }
    }

    public static TransferMessage valueOf(Object info, Map<String, Object> attachment) {
        CompositeByteBuf messageBuf = Unpooled.compositeBuffer();

        // header
        ByteBuf header = Unpooled.buffer();
        writeHeader(header, (short) 0, MessageType.valueOf(info.getClass()));
        messageBuf.addComponent(true, header);

        // messageBody
        messageBuf.addComponent(true, Unpooled.wrappedBuffer(TransferMessage.serialize(info)));

        // attachment
        writeAttachment(messageBuf, header, attachment);
        // reset total size
        header.setInt(HEADER_OFFSET[HEADER_TOTALLEN_IDX], messageBuf.readableBytes());

        return new TransferMessage(messageBuf);
    }

    private static <T> T readObject(ByteBuf buf, int len, Class<T> cls) {
        byte[] dst = new byte[len];
        buf.readBytes(dst);
        return TransferMessage.deserialize(dst, cls);
    }

    public static <T> T readMessageBody(ByteBuf messageBody, Class<T> clazz) {
        return TransferMessage.readObject(messageBody, messageBody.readableBytes(), clazz);
    }

    public static void release(Object obj, int hopeTime) {
        if (!(obj instanceof ReferenceCounted)) {
            return;
        }
        int idx = 0;
        while (true) {
            try {
                if (!ReferenceCountUtil.release(obj)) {
                    idx++;
                } else if (ReferenceCountUtil.refCnt(obj) == 0) {
                    break;
                }
            } catch (Exception e) {
                if (idx > hopeTime) {
                    logger.error("{}期望释放{}次，实际释放{}次", obj.getClass().getName(), hopeTime, idx);
                }
                return;
            }
        }
    }

}
